package com.dx.logic.service;

import com.dx.core.KafkaMessageVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 业务端mq监听
 *
 * @author xufeng
 * @since 04.04.2023
 */
@Slf4j
@Component
public class KafkaListenConsumer {

    @Autowired
    private LogicConsumer logicConsumer;

    /**
     * im server服务端协议监听
     *
     * @param records 消费信息
     * @param ack     Ack机制
     */
    // @KafkaListener(topics = "kafka4")
    public void consumerListen(List<ConsumerRecord<String, KafkaMessageVo>> records, Acknowledgment ack) {
        try {

            for (ConsumerRecord record : records) {
                log.info("---消费分区信息:{}", record.toString());
                //具体service里取做逻辑
                logicConsumer.processAction(record);
            }

        } catch (Exception e) {
            log.error("----数据消费者解析数据异常:{}", e.getMessage(), e);
        } finally {
            //手动提交偏移量
            ack.acknowledge();
        }
    }


}
